package net.bwie.realtime.vehicle.dwd.job.one;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.utils.DateTimeUtil;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.AreaControl;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

/*
 *  todo 1 实时车辆分布情况
 *   是指在一段时间内（比如：10分钟）整个城市中每个区分布多少量车。
 *   这里要注意车辆的去重，因为在10分钟内一定会有很多的车，经过不同的卡口。这些车牌相同的车，我们只统计一次。其实就是根据车牌号去重。
 * */

/**
 * @author ASUS
 */
public class VehicleDwsJob {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 增加网络和检查点配置
        env.setParallelism(1);
        // 增加检查点间隔
        env.enableCheckpointing(5000L);
        // 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // 设置网络缓冲区
        // 减少缓冲区超时时间，提高实时性
        env.setBufferTimeout(100);
        DataStream<String> vehicleKafkaStream = KafkaUtil.consumerKafka(env, "dwd_car_monitor");

        vehicleKafkaStream.print();
        //代码实现
        DataStream<String> result = mapWindowData(vehicleKafkaStream);
//        result.print();


        DorisUtil.saveToDoris(result,
                "vehicle_realtime_report",
                "dws_vehicle_car_report");

        env.execute("VehicleDwsJob");

    }

    private static DataStream<String> mapWindowData(DataStream<String> vehicleKafkaStream) {

        //转为实体类，并对时间戳进行毫秒级的转化
        SingleOutputStreamOperator<MonitorInfo> monitorinfo = vehicleKafkaStream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                jsonObject.put("actionTime", jsonObject.getLong("actionTime") * 1000);
                return JSON.parseObject(jsonObject.toJSONString(), MonitorInfo.class);
            }
        });
//        monitorinfo.print();

        //设置事件时间水位线
        DataStream<MonitorInfo> timeStream = monitorinfo.assignTimestampsAndWatermarks(
                WatermarkStrategy.<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(
                                (element, recordTimestamp) -> element.getActionTime()
                        )
        );

        //根据卡口进行分组进行去重
        KeyedStream<MonitorInfo, String> keyedStream = timeStream.keyBy(
                MonitorInfo::getAreaId
        );
//        keyedStream.print("1");

        SingleOutputStreamOperator<AreaControl> apply = keyedStream
                .window(TumblingEventTimeWindows.of(Time.minutes(10)))
                .apply(new WindowFunction<MonitorInfo, AreaControl, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<MonitorInfo> iterable, Collector<AreaControl> collector) throws Exception {


                        Set<String> set = new HashSet<String>();
                        for (MonitorInfo monitorInfo : iterable) {
                            set.add(monitorInfo.getCar());
                        }
                        String start = DateTimeUtil.convertLongToString(timeWindow.getStart(), "yyyy-MM-dd HH:mm:ss");
                        String end = DateTimeUtil.convertLongToString(timeWindow.getEnd(), "yyyy-MM-dd HH:mm:ss");
                        collector.collect(new AreaControl(s, set.size(), start, end));

                    }
                });
//        apply.print("PPP");


//        // 修改为输出JSON格式

        return apply.map(data -> {
            return String.format(
                    "{\"window_start\":\"%s\",\"window_end\":\"%s\",\"cur_date\":\"%s\",\"area_id\":\"%s\",\"vehicle_count\":%d}",
                    data.getWindowStart(),
                    data.getWindowEnd(),
                    data.getWindowEnd().split(" ")[0],
                    data.getAreaId(),
                    data.getVehicleCount()
            );
        });
    }
}
